package tcpsubpub

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/quewriter"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/kplbase/core/knet"
	"net"
	"sync"
	"sync/atomic"
)

type TcpServerNode struct {
	sessionSN   int32
	connectcnt  int32
	startedflag int32
	network     string
	addr        string
	maxonline   int
	serv        *knet.NetListener
	ex          *subpub.Subchannel
}

func (this *TcpServerNode) checkStart() error {
	if !atomic.CompareAndSwapInt32(&this.startedflag, 0, 1) {
		return nil
	}
	serv := knet.NewNetListener()
	serv.Config(this.network, this.addr)
	serv.OnErr = func(err error) {
		this.ex.Pub("evt", 0, subpub.ACTION_ERROR, err)
	}
	serv.OnListenSucc = func(network, addr string) {
		this.ex.Pub("evt", 0, subpub.ACTION_MESSAGE, fmt.Sprintf("%s:%s启动成功", network, addr))
	}
	serv.HandleConnFunc = func(conn net.Conn) {
		go this.innerHandleConnFunc(conn)
	}
	this.serv = serv

	serv.CheckStart()
	return nil
}

func (this *TcpServerNode) Close() error {
	serv := this.serv
	if serv != nil {
		serv.Close()
	}
	this.ex.Pub("notify-closed", 0, "请求关闭")
	return nil
}

func (this *TcpServerNode) PubBuf(max int, data []byte) int {
	return this.ex.Pub("send", max, data)
}

func (this *TcpServerNode) innerHandleConnFunc(conn net.Conn) {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}

	sessionid := fmt.Sprintf("%d-%s", atomic.AddInt32(&this.sessionSN, 1), gobase.NowIDString())
	cnt := atomic.AddInt32(&this.connectcnt, 1)
	if this.maxonline > 0 && cnt > int32(this.maxonline) {
		defer atomic.AddInt32(&this.connectcnt, -1)
		conn.Close()
		return
	}
	sublst := make(map[string]int8)

	defer func() {
		for k, _ := range sublst {
			this.ex.Unsub(sessionid, k)
		}
		defer atomic.AddInt32(&this.connectcnt, -1)
	}()

	var lk sync.RWMutex
	var quew *quewriter.QueueWriter
	defer func() {
		if quew != nil {
			lk.Lock()
			defer lk.Unlock()
			quew.Close()
		}
	}()

	var closedflag int32 = 0
	var reason string = ""

	var reqDis = func(s string) {
		if atomic.CompareAndSwapInt32(&closedflag, 0, 1) {
			reason = s
		}
		conn.Close()
	}

	var checkQueW = func() (w *quewriter.QueueWriter) {
		if atomic.LoadInt32(&closedflag) == 1 {
			return nil
		}
		lk.RLock()
		w = quew
		lk.RUnlock()
		if w != nil {
			return w
		}
		lk.Lock()
		defer lk.Unlock()
		w = quew
		if w == nil {
			quew = quewriter.NewQueueWriter(conn, 2048)
			quew.OnAfterWrite = func(p []byte, n int, err error) {
				if err == nil {
					this.ex.Pub("evt", 0, subpub.ACTION_SERVER_CONN_SEND_BUF, sessionid, p[:n])
				} else {
					reqDis(fmt.Sprintf("发送数据异常:%s", err.Error()))
				}
			}
			w = quew
		}
		return w
	}

	this.ex.Pub("evt", 0, subpub.ACTION_SERVER_CONN_CONNECTED, sessionid, conn.RemoteAddr().String())

	sublst["send"] = 1
	this.ex.Sub(sessionid, "send", func(id, topic string, args ...interface{}) (ok bool) {
		if len(args) == 0 {
			return false
		}
		buf := args[0].([]byte)
		if atomic.LoadInt32(&closedflag) == 0 {
			w := checkQueW()
			if w == nil {
				return false
			}
			w.Write(buf)
			return true
		}
		return false
	})

	sublst["notify-closed"] = 1
	this.ex.Sub(sessionid, "notify-closed", func(id, topic string, args ...interface{}) (ok bool) {
		if len(args) > 0 {
			reason, _ = args[0].(string)
		}
		if len(reason) == 0 {
			reason = "请求关闭(notify-closed)"
		}
		reqDis(reason)
		return true
	})

	for {
		buf := make([]byte, 4096)
		n, e1 := conn.Read(buf)
		if e1 != nil {
			if atomic.CompareAndSwapInt32(&closedflag, 0, 1) {
				reason = e1.Error()
			}
			break
		}
		this.ex.Pub("evt", 0, subpub.ACTION_SERVER_CONN_RECV_BUF, sessionid, buf[:n])
	}
	atomic.StoreInt32(&closedflag, 1)
	this.ex.Pub("evt", 0, subpub.ACTION_SERVER_CONN_DISCONNECTED, sessionid, reason)
}

var (
	DefaultTcpServerGroup = gobase.NewConnectObjectRefStorage(newTcpServerNode)
)

func newTcpServerNode(connstr string) interface{} {
	conf := gobase.NewStrMap()
	conf.URLFormDecode(connstr)
	node := &TcpServerNode{network: conf.StringByName("network", "tcp4"),
		addr: conf.StringByName("addr", "")}
	node.ex = subpub.NewSubchannel()
	return node
}
